package cn.zhang.violet.report

import cn.zhang.violet.config.ConfigHandler
import cn.zhang.violet.utils.{Jpools, RptKpi}
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

object TerminalRadioRedis {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("媒体分析--获取动态数据")
    conf.setMaster("local[*]")
    //设置spark程序采用的是序列化方式
    conf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(conf)

    val sQLContext = new SQLContext(sc)

    val dataFrame = sQLContext.read.parquet(ConfigHandler.parquetFilePath)

    //设置报表的维度
    dataFrame.mapPartitions(iter => {
      // 获取一个连接
      val jedis = Jpools.getJedis
      val result = iter.map(str => {
        // 维度字段
        var appName = str.getAs[String]("appname")
        val appId = str.getAs[String]("appid")

        if (StringUtils.isEmpty(appName)) { // appname为空了
          if (StringUtils.isNotEmpty(appId)) { // 去字典文件中找
            appName = jedis.hget("appdict", appId) //MySQL||Redis
            if (StringUtils.isEmpty(appName)) appName = "未知"
          } else {
            appName = "未知"
          }
        }
        (appName, RptKpi(str))
      })
      jedis.close()
      result
    }).reduceByKey((a,b) =>a.zip(b).map(t => t._1 + t._2))
        .sortBy(tp => tp._2.head,false)
      .saveAsTextFile("G:\\data\\log\\app1")



    sc.stop()
  }
}
